package com.atguigu.flink.timeAndwindow;

import com.atguigu.flink.func.ClickSource;
import com.atguigu.flink.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.HashSet;
import java.util.Set;

/**
 * Created by 黄凯 on 2023/6/18 0018 14:25
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 */
public class Flink07_AggregateFunction {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<Event> ds = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );
        ds.print("input");

        //需求: 每10秒统计UV(独立访客数)
        ds.keyBy(
                event -> true
        ).window(
                TumblingEventTimeWindows.of(Time.seconds(10))
        ).aggregate(
                new AggregateFunction<Event, Set<String>, Integer>() {


                    /**
                     * 创建累加器对象
                     * @return
                     */
                    @Override
                    public Set<String> createAccumulator() {
                        return new HashSet<>();
                    }

                    /**
                     * 每来一条数据累加一次
                     * @param value The value to add
                     * @param accumulator The accumulator to add the value to
                     * @return
                     */
                    @Override
                    public Set<String> add(Event value, Set<String> accumulator) {
                        accumulator.add(value.getUser());
                        return accumulator;
                    }

                    /**
                     * 返回结果
                     * @param accumulator The accumulator of the aggregation
                     * @return
                     */
                    @Override
                    public Integer getResult(Set<String> accumulator) {
                        return accumulator.size();
                    }

                    /**
                     * 一般不用实现。
                     */
                    @Override
                    public Set<String> merge(Set<String> a, Set<String> b) {
                        return null;
                    }
                }
        ).print();

        env.execute();

    }

}
